package com.bdqn.mr.rj;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class ReduceJoinMapper extends Mapper<LongWritable, Text, Text, TableBean> {


    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 为了拿到分片信息
        // 当前 MapTask 分配到的 数据分片
        InputSplit inputSplit = context.getInputSplit();
        FileSplit fileSplit = (FileSplit) inputSplit;

        // 获取到 数据分片 所属文件的文件名称
        name = fileSplit.getPath().getName();
    }

    private String name;

    /**
     * 每一行数据调用一次
     *
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fields = line.split("\t");
        if (StringUtils.isNotEmpty(name)) {
            // 文件名称不为空
            if (name.contains("order")) {
                // 处理 order.txt 文件的数据分片
                // 1001	01	1
                // 封装 key
                pidText.set(fields[1]);

                // 封装 value
                TABLE_BEAN.setOrderId(fields[0]);
                TABLE_BEAN.setPid(fields[1]);
                TABLE_BEAN.setAmount(Integer.parseInt(fields[2]));
                TABLE_BEAN.setTableName("order");

                TABLE_BEAN.setpName("");

                context.write(pidText, TABLE_BEAN);
            }

            if (name.contains("pd")) {
                // 处理 pd.txt 文件的数据分片
                // 01	小米
                // 封装 key
                pidText.set(fields[0]);

                // 封装 value
                TABLE_BEAN.setPid(fields[0]);
                TABLE_BEAN.setpName(fields[1]);
                TABLE_BEAN.setTableName("pd");

                TABLE_BEAN.setOrderId("");
                TABLE_BEAN.setAmount(0);

                context.write(pidText, TABLE_BEAN);
            }
        }
    }

    public static final Text pidText = new Text();
    public static final TableBean TABLE_BEAN = new TableBean();
}
